package com.lee.message.core.transaction;

import com.lee.message.core.utils.BaseUtils;
import com.lee.message.core.aspectj.TransactionalMessageAspect;
import com.lee.message.core.entity.TransMsgInfo;
import com.lee.message.core.mapper.TransMsgInfoMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.StringUtils;

/**
 * 抽象监听
 */
@Slf4j
@RocketMQTransactionListener
public class CustomRocketMQLocalTransactionListener implements RocketMQLocalTransactionListener {

    private final static String TRANSACTION_ID = "rocketmq_TRANSACTION_ID";

    @Autowired
    private PlatformTransactionManager manager;

    @Autowired
    private TransMsgInfoMapper mapper;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        TransactionStatus status = null;
        try {
            String transactionId = getTransactionId(msg);
            log.info("执行本地事务："+transactionId);
            status = TransactionalMessageAspect.currentTransactionStatus();

            // 添加记录回查事务使用
            TransMsgInfo info = new TransMsgInfo();
            info.setMsgId(transactionId);
            info.setCreateTime(BaseUtils.getCurDateStr());
            mapper.insert(info);

            if(!status.isCompleted()){
                manager.commit(status);
            }

            return RocketMQLocalTransactionState.COMMIT;
        } catch (Throwable t) {
            log.error("执行本地事务失败", t);
            if(!status.isCompleted()){
                manager.rollback(status);
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        try {
            String transactionId = getTransactionId(msg);
            log.info("回查本地事务："+transactionId);
            // 判断本地事务是否已经提交
            int count = mapper.count(transactionId);
            if(count > 0){
                return RocketMQLocalTransactionState.COMMIT;
            }else{
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        } catch (Throwable t){
            log.error("回查本地事务失败", t);
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    private String getTransactionId(Message msg){
        Object object = msg.getHeaders().get(TRANSACTION_ID);
        if(object == null){
            throw new RuntimeException("消息"+TRANSACTION_ID+"属性不存在");
        }

        String transactionId = String.valueOf(object);
        if(!StringUtils.hasText(transactionId)){
            throw new RuntimeException("消息"+TRANSACTION_ID+"值不存在");
        }

        return transactionId;
    }
}
